package org.zjvis.datascience.service;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.zjvis.datascience.common.constant.SqlTemplate;
import org.zjvis.datascience.common.dto.PipelineInstanceDTO;
import org.zjvis.datascience.common.dto.TaskInstanceDTO;
import org.zjvis.datascience.common.enums.TaskInstanceStatus;
import org.zjvis.datascience.common.util.SqlUtil;
import org.zjvis.datascience.common.util.ToolUtil;
import org.zjvis.datascience.common.vo.DataPreviewVO;
import org.zjvis.datascience.service.dataprovider.GPDataProvider;
import org.zjvis.datascience.service.mapper.PipelineInstanceMapper;
import org.zjvis.datascience.service.mapper.TaskInstanceMapper;

/**
 * @description DataPreview 数据预览 Service
 * @date 2021-09-17
 */
@Service
public class DataPreviewService {

    private final static Logger logger = LoggerFactory.getLogger("DataPreviewService");

    @Autowired
    private PipelineInstanceMapper pipelineInstanceMapper;

    @Autowired
    private TaskInstanceMapper taskInstanceMapper;

    @Autowired
    private GPDataProvider gpDataProvider;

    public DataPreviewVO buildDataPreviewVO(String tableName, Long taskId, Long pipelineId,
        Long sessionId) {
        DataPreviewVO dataPreviewVO = new DataPreviewVO();
        dataPreviewVO.setTaskId(taskId);
        dataPreviewVO.setPipelineId(pipelineId);
        dataPreviewVO.setPipelineInstanceId(sessionId);
        if (tableName == null) {
            return null;
        }
        dataPreviewVO.setName(tableName);

        String[] tmps = tableName.split("\\.");
        // TODO
        Map<String, String> meta = new HashMap<>();
        if (tableName.startsWith(SqlTemplate.SOURCE_SCHEMA + ".")) {
            meta = gpDataProvider.getTableMetaMap(tmps[tmps.length - 1], SqlTemplate.SOURCE_SCHEMA);
        } else if (tableName.startsWith(SqlTemplate.SCHEMA + ".") || tableName.startsWith("solid_")
            || tableName.startsWith("view_")) {
            meta = gpDataProvider.getTableMetaMap(tmps[tmps.length - 1], SqlTemplate.SCHEMA);
        } else {
            meta = gpDataProvider.getTableMetaMap(tmps[tmps.length - 1], SqlTemplate.SOURCE_SCHEMA);
        }
        JSONArray head = new JSONArray();
        List<String> columns = new ArrayList<>();
        for (Map.Entry<String, String> entry : meta.entrySet()) {
            JSONObject item = new JSONObject();
            item.put("name", entry.getKey());
            item.put("type", SqlUtil.changeType(entry.getValue().toUpperCase()));
            head.add(item);
            columns.add(entry.getKey());
        }
        dataPreviewVO.setHead(head);
        Long count = gpDataProvider
            .getRecordCount(ToolUtil.alignTableName(tmps[tmps.length - 1], 0L));
        if (count == null) {
            logger.error("getRecordCount fail!!!!");
            return null;
        }
        dataPreviewVO.setCount(count);
        JSONArray dataArray = gpDataProvider
            .getPreviewData(ToolUtil.alignTableName(tmps[tmps.length - 1], 0L), columns, 20);
        if (dataArray == null) {
            logger.error("getPreviewData fail!!!1");
            return null;
        }
        dataPreviewVO.setData(dataArray);
        return dataPreviewVO;
    }

    public List<DataPreviewVO> preview(Long taskId, Long pipelineId) {
        List<DataPreviewVO> dataPreviewVOS = new ArrayList<>();
        PipelineInstanceDTO latestInstance = pipelineInstanceMapper.getLatestInstance(pipelineId);
        if (latestInstance == null) {
            logger.error("getLatestInstance fail!!!");
            return dataPreviewVOS;
        }

        Long sessionId = latestInstance.getId();

        Map<String, Long> paramMap = new HashMap<>();
        paramMap.put("sessionId", sessionId);
        paramMap.put("taskId", taskId);
        TaskInstanceDTO taskInstanceDTO = taskInstanceMapper.queryBySpecTaskInstance(paramMap);
        if (taskInstanceDTO == null) {
            logger.error("queryBySpecTaskInstance fail!!!!");
            return dataPreviewVOS;
        }
        String logInfo = taskInstanceDTO.getLogInfo();
        if (!TaskInstanceStatus.SUCCESS.toString().equals(taskInstanceDTO.getStatus())) {
            logger.debug("logInfo is null or status is not equal with success!!!");
            return new ArrayList<>();
        } else if (logInfo == null) {
            // 视图表预览处理
            String sqlText = taskInstanceDTO.getSqlText();
            if (StringUtils.isEmpty(sqlText) || !sqlText.toUpperCase().startsWith("CREATE VIEW")) {
                logger.error("sqlText is empty or sqlText is not view!!!");
                return new ArrayList<>();
            }
            String tableName = sqlText.split(" ")[2];
            DataPreviewVO vo = this.buildDataPreviewVO(tableName, taskId, pipelineId, sessionId);
            vo.setCreateTime(taskInstanceDTO.getGmtCreate());
            dataPreviewVOS.add(vo);
            return dataPreviewVOS;
        }

        JSONObject jsonObject;
        try {
            jsonObject = JSONObject.parseObject(logInfo);
            JSONObject result = jsonObject.getJSONObject("result");
            if (!result.containsKey("output_params")) {
                logger.error("output_params is not exist!!!");
                return null;
            }
            JSONArray outputs = result.getJSONArray("output_params");
            for (int i = 0; i < outputs.size(); ++i) {
                JSONObject obj = outputs.getJSONObject(i);
                String tableName = obj.getString("out_table_name");
                DataPreviewVO vo = this
                    .buildDataPreviewVO(tableName, taskId, pipelineId, sessionId);
                vo.setCreateTime(taskInstanceDTO.getGmtCreate());
                dataPreviewVOS.add(vo);
            }
        } catch (Exception e) {
            logger.error(e.getMessage());
            return null;
        }
        return dataPreviewVOS;
    }

}
